package com.idroc.coelacanth.service;

import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.idroc.coelacanth.core.membership.Endpoint;
import com.idroc.coelacanth.core.message.Message;
import com.idroc.coelacanth.core.message.MessageFactory;

/**
 * 消息服务
 * @author yuyoo (yuyoo4j@163.com)
 * @date Dec 9, 2009 10:55:11 PM
 */
public class MessageService {

	private static final Logger log = LoggerFactory.getLogger(MessageService.class);
	
	/**
	 * 端点, 消息服务端建立在此端点上
	 */
	private Endpoint ep = null;
	
	/**
	 * 消息服务端Bootstrap
	 */
	private ServerBootstrap serverBootstrap = null;
	
	/**
	 * 消息服务端工厂
	 */
	private MessageFactory serverMsgFactory = null;
	
	/**
	 * 消息客户端Bootstrap
	 */
	private ClientBootstrap clientBootstrap = null;
	 
	/**
	 * 消息客户端工厂
	 */
	private MessageFactory clientMsgFactory = null;
	
	/**
	 * 端点连接管理器
	 */
	private EndpointConnectionManager connectionManager = null;
	
	public MessageService(Endpoint ep) {
		
		this.ep = ep;
	}
	
	/**
	 * 返回服务端消息工厂
	 * @return -- 消息工厂
	 */
	public MessageFactory getServerMessageFactory() {
		
		return serverMsgFactory;
	}
	
	/**
	 * 返回客户端消息工厂
	 * @return --  消息工厂
	 */
	public MessageFactory getClientMessageFactory() {
		
		return clientMsgFactory;
	}
	
	/**
	 * 返回端点连接管理器
	 * @return -- 端点连接管理器
	 */
	public EndpointConnectionManager getEndpointConnectionManager() {
		
		return connectionManager;
	}
	
	/**
	 * 启动消息服务
	 */
	public void start() {
		
		startMessageServerService();
		startMessageClientService();
		connectionManager = new EndpointConnectionManager();
	}
	
	/**
	 * 启动消息服务端服务
	 */
	private void startMessageServerService() {
		
		ChannelFactory factory = new NioServerSocketChannelFactory (
										Executors.newCachedThreadPool(),
										Executors.newCachedThreadPool());

		serverBootstrap = new ServerBootstrap (factory);
		
		serverMsgFactory = new MessageFactory();
		 
		
		serverBootstrap.setPipelineFactory(serverMsgFactory);
		serverBootstrap.setOption("child.tcpNoDelay", true);
		serverBootstrap.setOption("child.keepAlive", true);
		if (null == ep.getHost()) {
			serverBootstrap.bind(new InetSocketAddress(ep.getPort()));
		} else {
			serverBootstrap.bind(new InetSocketAddress(ep.getHost(), ep.getPort()));
		}
		
	}
	
	/**
	 * 启动消息客户端服务
	 */
	private void startMessageClientService() {
		
		ChannelFactory factory = new NioClientSocketChannelFactory (
										Executors.newCachedThreadPool(),
										Executors.newCachedThreadPool());

		clientBootstrap = new ClientBootstrap(factory);
		
		clientMsgFactory = new MessageFactory();
				
		clientBootstrap.setPipelineFactory(clientMsgFactory);
		
		clientBootstrap.setOption("tcpNoDelay" , true);
		clientBootstrap.setOption("keepAlive", true); 
	}
	
	/**
	 * 停止消息服务
	 */
	public void stop() {
		
	}

	/**
	 * 向节点发送消息
	 * @param ep -- 节点 
	 * @param msg -- 消息
	 * @return -- 通道结果 ChannelFuture
	 */
	public ChannelFuture sendMessage(final Endpoint ep, final Message msg) {
		
		final long current = System.currentTimeMillis();
		log.debug("发送任务[{}]向端点[{}]发送消息", current, ep);
		log.debug("发送消息[{}]", msg);
		
		try {
			
			Channel ch = connectionManager.getConnection(ep); 
			ChannelFuture cf = ch.write(msg);
			cf.addListener(new ChannelFutureListener() {
	
				public void operationComplete(ChannelFuture writeFuture)
						throws Exception { 
	
					if (!writeFuture.isSuccess()) {
						log.error("发送任务[{}]写消息[{}]数据异常", current, msg);
						log.error("任务发送异常,将立即关闭端点通道", writeFuture.getCause());
						connectionManager.closeConnection(ep); 
					} else {
						log.debug("发送任务[{}]完成", current);
					}
				} 
			}); 
			return cf;
		} catch (RuntimeException rex) {
			String errorMsg = "发送任务[" + current + "]为端点[" + ep + "]创建通道异常";
			log.error(errorMsg); 
			log.error("发送任务[{}]将立即关闭端点通道", current);
			connectionManager.closeConnection(ep); 
			throw rex;
		}
		 
	}
	
	/**
	 * 端点连接管理器
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date Dec 9, 2009 10:56:00 PM
	 */
	public class EndpointConnectionManager {
		
		/**
		 * 已经建立通道的连接集合
		 */
		private final Map<Endpoint, Channel> channels = new HashMap<Endpoint, Channel>(200);
		
		/**
		 * 建立通道超时时间
		 */
		private final long buildConnectionTimeout = 6000; // 6s
		
		private EndpointConnectionManager() {}
		
		/**
		 * 获取端点的通道
		 * @param ep -- 端点
		 * @return -- 通道
		 */
		public Channel getConnection(Endpoint ep) {
			
			Channel ch = channels.get(ep);
			if (null == ch) {
				ch = buildConnection(ep);
				registerConnetion(ep, ch);
			}
			return ch;
		}
		
		/**
		 * 关闭端点的通道
		 * @param ep  -- 端点
		 * @return -- 通道结果ChannelFuture
		 */
		public ChannelFuture closeConnection(Endpoint ep) {
			
			Channel ch = channels.remove(ep);
			if (null != ch) {
				return ch.close();
			}
			return null;
		}
		
		/**
		 * 注册端点的通道到管理器
		 * @param ep -- 端点
		 * @param ch -- 通道
		 */
		public void registerConnetion(final Endpoint ep, final Channel ch) {
			
			channels.put(ep, ch);
		}
		
		/**
		 * 创建通道
		 * @param ep -- 端点
		 * @return -- 通道
		 */
		public Channel buildConnection(final Endpoint ep) {
			

			final long current = System.currentTimeMillis();
			log.debug("连接任务[{}]为端点[{}]创建通道", current, ep); 
			
			final BlockingQueue<Object> fetchQueue = new ArrayBlockingQueue<Object>(1); 
			ChannelFuture cf = clientBootstrap.connect(new InetSocketAddress(ep.getHost(), ep.getPort()));
			cf.addListener(new ChannelFutureListener() {
	
				public void operationComplete(ChannelFuture openFuture)
						throws Exception { 
					
					if (openFuture.isSuccess()) {
						log.debug("连接任务[{}]创建通道成功", current); 
						fetchQueue.offer(openFuture.getChannel());
					} else {
						log.error("连接任务[{}]为端点[{}]创建通道异常", current, ep); 
						fetchQueue.offer(openFuture.getCause());
					}    
				} 
			}); 
			Object rtv = null;
			try { 
				rtv = fetchQueue.poll(buildConnectionTimeout, TimeUnit.MILLISECONDS);
			} catch (InterruptedException ex) { 
				String errorMsg = String.format("连接任务[{}]创建通道超时", current);
				log.error(errorMsg);
				throw new RuntimeException(errorMsg, ex);
			}

			if (rtv instanceof Channel) {
				return (Channel)rtv;
			} else {
				throw new RuntimeException((Throwable)rtv);
			}
		}
		
		/**
		 * 异步创建通道
		 * @param ep -- 端点
		 * @return -- 通道结果ChannelFuture
		 */
		public ChannelFuture asyncBuildAndRegisterConnection(final Endpoint ep) {
			 
			final long current = System.currentTimeMillis();
			log.debug("连接任务[{}]为端点[]异步创建通道", current, ep); 
			
			ChannelFuture cf = clientBootstrap.connect(new InetSocketAddress(ep.getHost(), ep.getPort()));
			cf.addListener(new ChannelFutureListener() {
	
				public void operationComplete(ChannelFuture openFuture)
						throws Exception { 
					
					if (openFuture.isSuccess()) {
						log.debug("连接任务[{}]为端点[{}]异步创建通道", current, ep);  
						registerConnetion(ep, openFuture.getChannel());
					} else {
						log.error("连接任务[{}]为端点[{}]异步创建通道异常", current, ep);  
					}    
				} 
			}); 
			return cf;
		}
	}
}
